package hang.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * @author 航
 * @date 2024/04/14 12:31
 **/
public class ProducerPartitionerTest {
    public static void main(String[] args) {
        // 创建 配置对象
        Map<String, Object> config = new HashMap<String, Object>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 对生产者的数据 K，V 进行序列化操作
        // StringSerializer.class.getName() 表示获取这个类的名称
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 分区器
        config.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, new MyPartitioner());


        // 创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer(config);

//        // 创建数据
//        // 构建数据，需要传递三个参数：topicName，数据的key，数据的value
//        ProducerRecord<String, String> record = new ProducerRecord("test","key","value");
//
//        // 通过生产者对象发送数据
//        producer.send(record);

        for(int i = 0; i<10; i++){
            ProducerRecord<String, String> record = new ProducerRecord("test","key" + i,"value" + i);
            producer.send(record);
        }

        // 关闭生产者对象
        producer.close();
    }
}
